package cn.ucox.web.ms.monitor;

import cn.ucox.web.ms.components.ftp.DexFtpClient;
import cn.ucox.web.ms.config.DataExchangeConfig;
import cn.ucox.web.ms.redis.RedisClient;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.pool2.PooledObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.*;
import java.util.stream.Collectors;

import static cn.ucox.web.ms.Constant.KEY_FTP_FILE_DETENTION_NUM;
import static cn.ucox.web.ms.Constant.SYNC_ANCHOR;
import static cn.ucox.web.ms.monitor.SyncFileParser.FTP_READY_TRANS_DATA;
import static cn.ucox.web.ms.monitor.SyncFileParser.SYNC_FTP_FILE_INDEX;

/**
 * FTP同步目录监控
 *
 * @author chenw
 * @create 2017-05-26 21:20
 * @email javacspring@gmail.com
 */
public class FtpDirectoryListener {

    private static final Logger logger = LoggerFactory.getLogger(FtpDirectoryListener.class);

    private static final String FTP_TRANS_DATA_ANCHOR_PREFIX = "SYNC:FTP:DATA:ANCHOR";
    public static final String FTP_TRANS_DATA_TIMER = "FTP:TRANS:DATA:TIMER";//事务数据记时
    public static final String FTP_SYNC_ANCHOR = "SYNC:FTP:MONITOR";
    public static final String LAST_SYNC_TIME = "lst";
    public static final String LAST_SYNC_DIRECTORY_FILES_COUNT = "lsdfc";
    public static final String LAST_SYNC_FILE_NAME = "lsfn";
    public static final String LAST_SYNC_FILE_CREATE_TIME = "lsfct";
    public static final String LAST_SYNC_TRANS_ID = "lsti";
    public static final String LAST_SYNC_TRANS_DATA_SIZE = "lstds";
    public static final String LAST_SYNC_TRANS_DATA_FRAGMENTS = "lstdf";
    public static final String LAST_SYNC_CURR_DATA_FRAGMENT = "lscdf";

    private DexFtpClient dexFtpClient;
    private DataExchangeConfig.PollingConfig pollingConfig;

    private FtpDirectoryMonitor directoryMonitor;
    private RedisClient redisClient = RedisClient.instance();
    private boolean exit = false;
    private DecimalFormat fragFormat = new DecimalFormat("00000");


    public FtpDirectoryListener(DexFtpClient dexFtpClient, DataExchangeConfig.PollingConfig pollingConfig) {
        this.dexFtpClient = dexFtpClient;
        this.directoryMonitor = new FtpDirectoryMonitor();
        this.pollingConfig = pollingConfig;
    }

    /**
     * 开始监听FTP文件目录
     */
    public void start() {
        //初始化FTP监控状态表
        initFTPMonitor();
        directoryMonitor.start();
    }

    public void initFTPMonitor() {
        redisClient.hmsetx(FTP_SYNC_ANCHOR, LAST_SYNC_DIRECTORY_FILES_COUNT, "0", LAST_SYNC_TIME, "0", LAST_SYNC_FILE_NAME, "", LAST_SYNC_FILE_CREATE_TIME, "0", LAST_SYNC_TRANS_ID, "0", LAST_SYNC_TRANS_DATA_SIZE, "0", LAST_SYNC_TRANS_DATA_FRAGMENTS, "0", LAST_SYNC_CURR_DATA_FRAGMENT, "0");
    }

    public void stop() {
        exit = true;
        logger.info("FTP目录监控组件停止");
    }


    /**
     * 远程FTP文件目录状态监控
     */
    class FtpDirectoryMonitor extends Thread {
        @Override
        public void run() {
            //轮询FTP目录状态
            while (true) {
                if (exit) break;
                Map<String, String> tranDataAnchor;
                Jedis jedis = null;
                try {
                    //检测FTP目录增量文件,避免重复读取FTP文件，防止分片数据计数错误
                    /*List<FTPFile> incrementFiles = getIncrementFiles(); //已将FTP目录中的文件添加至REDIS中
                    if (null == incrementFiles || incrementFiles.isEmpty()) {
                        delayTime();
                        continue;
                    }*/

                    List<String> incrementFileNames = getIncrementFileNames(); //已将FTP目录中的文件添加至REDIS中
                    if (null == incrementFileNames || incrementFileNames.isEmpty()) {
                        delayTime();
                        continue;
                    }

                    //预处理FTP文件数据
                    jedis = redisClient.getJedis();
                    Pipeline pipeline = jedis.pipelined();
                    for (String fileName : incrementFileNames) {
                        //提取文件名称，并根据事务ID+分片ID作为score
                        String[] info = fileName.substring(0, fileName.lastIndexOf(".")).split("_");
                        if (4 != info.length) {
                            logger.error("事务分片数据文件名{}错误", fileName);
                            continue;
                        }
                        String transId = info[0];//事务编号
                        String fragNum = info[1];//总分片数量
                        String dataSize = info[3];//事务数据大小

                        //加载FTP事务数据锚点信息
                        List<String> resp = jedis.hmget(getTransactionDataAnchorKey(transId), "trans", "size", "frags", "received", "startScore", "endScore");
                        if (null == resp || null == resp.get(0)) { //没有缓存记录
                            tranDataAnchor = new HashMap<String, String>(6) {{
                                put("trans", transId);
                                put("size", dataSize);
                                put("frags", fragNum);
                                put("received", "1");
                                put("startScore", transId + fragFormat.format(1));
                                put("endScore", transId + fragFormat.format(Long.valueOf(fragNum)));
                            }};
                        } else {
                            tranDataAnchor = new HashMap<String, String>(6) {{
                                put("trans", resp.get(0));
                                put("size", resp.get(1));
                                put("frags", resp.get(2));
                                put("received", String.valueOf(Integer.valueOf(resp.get(3)) + 1));
                                put("startScore", fragFormat.format(Long.valueOf(resp.get(4))));
                                put("endScore", fragFormat.format(Long.valueOf(resp.get(5))));
                            }};
                        }
                        //如果已接收到事务数据分片与实际已存在数据分片值相同，则标记此事务数据可以进行解析
                        if (Objects.equals(fragNum, tranDataAnchor.get("received"))) {
                            //将已标记完成事务数据锚点信息进行排序
                            jedis.zadd(FTP_READY_TRANS_DATA, Long.valueOf(transId), String.format("%s %s", tranDataAnchor.get("startScore"), tranDataAnchor.get("endScore")));
                            //同步删除redis锚点缓存信息
                            jedis.del(getTransactionDataAnchorKey(transId));
                            jedis.hdel(FTP_TRANS_DATA_TIMER, transId);//如果数据已完整，则从计时器中删除
                            logger.debug("事务{}数据已就绪,开始解析处理", transId);
                        } else {
                            jedis.hmset(getTransactionDataAnchorKey(transId), tranDataAnchor);
                            //刷新最后一次事务数据激活时间时间(事务中每个分片数据均会激活一次)
                            jedis.hset(FTP_TRANS_DATA_TIMER, transId, String.valueOf(System.currentTimeMillis()));
                        }
                    }
                    pipeline.sync();
                } catch (Exception ex) {
                    logger.error("FTP同步文件处理错误:{}", ex.getMessage(), ex);
                } finally {
                    if (null != jedis) {
                        try {
                            jedis.close();
                        } catch (Exception ex) {
                            logger.error("关闭jedis客户端错误", ex);
                        }
                    }
                }
            }
        }
    }

    private List<String> getIncrementFileNames() {
        PooledObject<FTPClient> pooledObject = null;
        List<String> newFileNames = null;
        try {
            pooledObject = dexFtpClient.get();
            if (null != pooledObject) {
                FTPClient ftpClient = pooledObject.getObject();
                ftpClient.enterLocalPassiveMode();
                String[] fileNames = ftpClient.listNames();
                Jedis jedis = null;
                try {
                    jedis = redisClient.getJedis();
                    if (null == jedis) {
                        logger.error("无法获取Jedis实例");
                        return new ArrayList<>(0);
                    }
                    redisClient.hset(SYNC_ANCHOR, KEY_FTP_FILE_DETENTION_NUM, String.valueOf(fileNames.length));
                    if (0 == fileNames.length) {
                        logger.info("-FTP监控目录滞留文件:0,增量数据:无");
                        return new ArrayList<>(0);
                    }

                    // 检测SYNC:FTP:FILES中条数,超过50，等待
                    long indexFiles = jedis.zcard(SYNC_FTP_FILE_INDEX);
                    if (indexFiles >= pollingConfig.getMaxFtpDetention()) {
                        logger.info("-就绪文件队列数超出最大值:{}:{}", SYNC_FTP_FILE_INDEX, pollingConfig.getMaxFtpDetention());
                        return null;
                    }

                    Pipeline pipeline = jedis.pipelined();
                    Jedis finalJedis = jedis;

                    newFileNames = Arrays.stream(fileNames).filter(s -> {
                        if (!s.endsWith(".dex")) {
                            logger.error("数据文件名称不合法:{}", s);
                            return false;
                        }

                        String fileName = s;
                        String[] info = fileName.substring(0, fileName.lastIndexOf(".")).split("_");
                        if (4 != info.length) {
                            logger.error("事务分片数据文件名{}错误", fileName);
                            try {
                                ftpClient.deleteFile(s);
                            } catch (IOException e) {
                                logger.error("删除错误文件错误:{}", s, e);
                            }
                            return false;
                        }
                        if (Integer.valueOf(info[2]) > Integer.valueOf(info[1])) {
                            logger.error("事务分片数据文件{}错误,当前分片索引已超出总分片值", fileName);
                            try {
                                ftpClient.deleteFile(s);
                            } catch (IOException e) {
                                logger.error("删除错误文件错误:{}", s, e);
                            }
                            return false;
                        }

                        String score = info[0] + fragFormat.format(Integer.valueOf(info[2]));
                        return 1 == finalJedis.zadd(SYNC_FTP_FILE_INDEX, Long.valueOf(score), s);
                    }).collect(Collectors.toList());

                    //更新数据增量锚点文件滞留数
                    jedis.hset(SYNC_ANCHOR, KEY_FTP_FILE_DETENTION_NUM, String.valueOf(fileNames.length));
                    pipeline.sync();
                    if (null == newFileNames || 0 == newFileNames.size()) {
                        logger.info("FTP监控目录滞留文件:{},增量数据:无", fileNames.length);
                        return new ArrayList<>(0);
                    } else {
                        logger.info("FTP监控目录滞留文件:{},增量数据:{}", fileNames.length, newFileNames.size());
                    }
                } catch (Exception ex) {
                    logger.info("检测FTP监控目录增量文件错误:{}", ex.getMessage(), ex);
                    return new ArrayList<>(0);
                } finally {
                    if (jedis != null) {
                        jedis.close();
                    }
                }
            }
            return newFileNames;
        } catch (IOException e) {
            logger.error("检测FTP监控目录增量文件错误:{}", e.getMessage(), e);
            return new ArrayList<>(0);
        } finally {
            dexFtpClient.close(pooledObject);
        }
    }

    private void delayTime() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ex) {
            logger.error("延迟错误", ex);
        }
    }

    private String getTransactionDataAnchorKey(String transId) {
        return String.format("%s:%s", FTP_TRANS_DATA_ANCHOR_PREFIX, transId);
    }

}